package com.sf.kafka;

import java.util.Arrays;
import java.util.Iterator;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

public class KafkaConsumerClient<K,V> {
	private final KafkaConsumer<K, V> consumer;
	private final Properties props;
	
	public KafkaConsumerClient(String broker,String consumerGroup){
		this.props = new Properties();
		props.setProperty("bootstrap.servers", broker);
		props.setProperty("group.id", consumerGroup);
		//true时，Consumer会在消费消息后将offset同步到zookeeper，这样当Consumer失败后，新的consumer就能从zookeeper获取最新的offset
		props.setProperty("auto.commit.enable", "true");
		//自动提交的时间间隔
		props.setProperty("auto.commit.interval.ms", "1000");
		//zookeeper的心跳超时时间，超过这个时间就认为是无效的消费者
		props.setProperty("session.timeout.ms", "30000");
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		this.consumer = new KafkaConsumer<>(props);
	}
	
	/**
	 * 根据offset查询某个主题某个分区下的num条数据
	 * @param topicName:kafka主题
	 * @param partition:分区
	 * @param offset:偏移量
	 * @param num
	 */
	public void getRecordByOffset(String topicName,int partition,Long offset,int num) {
		TopicPartition part = new TopicPartition(topicName, partition);
		consumer.assign(Arrays.asList(part));
		consumer.seek(part, offset);
		ConsumerRecords<K, V> records = null;
		while(true) {
			records = consumer.poll(100);
			if(!records.isEmpty()) {
				break;
			}
		}
		System.out.println(records.count());
		Iterator<ConsumerRecord<K, V>> iterable = records.iterator();
		int index = 0;
		while(index<num && iterable.hasNext()) {
			System.out.println(iterable.next().toString());
			index++;
		}
	}
	
	/**
	 * 获取某个主题某个分区下的最后一个offset
	 * @param topicName
	 * @param partition
	 * @return
	 */
	public Long getLastOffset(String topicName,int partition) {
		TopicPartition part = new TopicPartition(topicName, partition);
		consumer.assign(Arrays.asList(part));
		consumer.seekToEnd(Arrays.asList(part));
		return consumer.position(part)-1;
	}
	
	/**
	 * 普通消费，依据zk记录offset,从offset开始消费
	 * @param topicName
	 */
	public void subscribe(String topicName) {
		consumer.subscribe(Arrays.asList(topicName));
		while(true) {
			ConsumerRecords<K, V> records = consumer.poll(100);
			for(ConsumerRecord record:records) {
				System.out.println(record.toString());
			}
		}
	}
	
	/**
	 * 从某个主题某个分区开始消费
	 * @param topicName
	 * @param partition
	 */
	public void subscribeFromBegin(String topicName,int partition) {
		TopicPartition part = new TopicPartition(topicName, partition);
		consumer.assign(Arrays.asList(part));
		consumer.seekToBeginning(Arrays.asList(part));
		while(true) {
			ConsumerRecords<K, V> records = consumer.poll(100);
			for(ConsumerRecord record:records) {
				System.out.println(record.toString());
			}
		}
	}
	
	public void close() {
		consumer.close();
	}
	public static void main(String[] args) {
		int len = args.length;
		String broker = "localhost:9092";
		String topic = "last-line-test";
		String consumerGroup = "test";
		int partiton = 0;
		long offset = 0L;
		int recordNum = 1;
		if(len >= 4) {
			broker = args[0];
			topic = args[1];
			consumerGroup = args[2];
			partiton = Integer.valueOf(args[3]);
		}
		if(len >=6) {
			offset = Long.valueOf(args[4]);
			recordNum = Integer.valueOf(args[5]);
		}
		KafkaConsumerClient<String, String> client = new KafkaConsumerClient<>(broker,consumerGroup);
		if(len == 4) {
			offset = client.getLastOffset(topic, partiton);
			System.out.println("=======================================================");
			System.out.println("topic:"+topic+",partiton:"+partiton+",endOffset:"+offset);
			System.out.println("=======================================================");
		}else if(len == 6){
			client.getRecordByOffset(topic, partiton, offset,recordNum);
			/*try {
				Thread.sleep(3000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}*/
		}else {
			System.err.println("参数错误");
		}
		client.close();
	}

}
